package com.hzgj.bcl.soa.rpc.client;

import com.hzgj.bcl.soa.Address;
import com.hzgj.bcl.soa.rpc.RequestCorrelator;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Rsp;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class RpcRequest<T> extends Request {
    protected final Rsp<T> result;
    protected int num_received = 0;


    public RpcRequest(Message msg, RequestCorrelator corr, RequestOptions options) {
        super(msg, corr, options);
        result = new Rsp<>();
    }

    @Override
    protected void sendRequest() throws Exception {
        try {
            if (log.isTraceEnabled()) {
                log.trace(new StringBuilder("sending request (id=").append(req_id).append(')')
                        .toString());
            }
            corr.sendRequest(req_id, request_msg, this);
        } catch (Exception ex) {
            if (corr != null) {
                corr.done(req_id);
            }
            throw ex;
        }
    }


    /* ---------------------- Interface RspCollector -------------------------- */

    /**
     * <b>Callback</b> (called by RequestCorrelator or Transport).
     * Adds a response to the response table. When all responses have been received,
     * <code>execute()</code> returns.
     */
    @Override
    public void receiveResponse(Object response_value, Address sender, boolean is_exception) {

        lock.lock();
        try {
            if (done) {
                return;
            }
            if (!result.wasReceived()) {
                num_received++;
                if (is_exception && response_value instanceof Throwable) {
                    result.setException((Throwable) response_value);
                } else {
                    result.setValue((T) response_value);
                }
            }
            done = responsesComplete();
            if (done && corr != null) {
                corr.done(req_id);
            }
        } finally {
            cond.signal(true); // wakes up execute()
            lock.unlock();
        }
        checkCompletion(this);
    }

    public boolean responseReceived() {
        return num_received >= 1;
    }


    /* -------------------- End of Interface RspCollector ----------------------------------- */

    public Rsp<T> getResult() {
        return result;
    }


    public T getValue() throws ExecutionException {
        if (result.hasException()) {
            throw new ExecutionException(result.getException());
        }

        return result.getValue();
    }


    @Override
    public T get() throws InterruptedException, ExecutionException {
        lock.lock();
        try {
            waitForResults(0);
            return getValue();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        boolean ok;
        lock.lock();
        try {
            ok = waitForResults(unit.toMillis(timeout));
        } finally {
            lock.unlock();
        }
        if (!ok) {
            throw new TimeoutException();
        }
        return getValue();
    }

    @Override
    protected boolean responsesComplete() {
        return done || result.wasReceived() || result.wasUnreachable() || num_received >= 1;
    }
}